package mq.cinrad.kafka.app.producer;

import java.io.File;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;

public class HttpFileActionCallback implements FileActionCallback {

	private static final Logger logger = LoggerFactory.getLogger(HttpFileActionCallback.class);

	private KafkaTemplate<String, String> template;

	private KafkaWatchFileFilter kafkaWatchFileFilter;

	public HttpFileActionCallback(KafkaTemplate<String, String> template, KafkaWatchFileFilter filter) {

		this.template = template;
		this.kafkaWatchFileFilter = filter;

	}

	@Override
	public void delete(File file) {

		if (null != file && null != file.getAbsolutePath()) {
			String fileUri = file.toURI().toString();
			logger.debug("delete file: {}", fileUri);
		}

	}

	@Override
	public void modify(File file) {
		if (null != file && null != file.getAbsolutePath()) {
			String fileUri = file.toURI().toString();
			logger.debug("modify file: {}", fileUri);
		}

	}

	@Override
	public void create(File file) {

		if (null != file && null != file.getAbsolutePath()) {
			// String fname = file.getAbsolutePath();
			String fileUri = file.toURI().toString();
			logger.debug("create file: {}", fileUri);

			if (file.exists() && !file.isDirectory()
					&& kafkaWatchFileFilter.accept(file.getParentFile(), file.getName())) {

				String topic = kafkaWatchFileFilter.getKafkaWatchFileConfig().getDirTopicBiMap()
						.get(file.getParentFile().getAbsolutePath());

				if (topic != null) {

					String httpFileUri = fileUri;
					if (kafkaWatchFileFilter.getKafkaWatchFileConfig().getNameRegex() != null
							&& kafkaWatchFileFilter.getKafkaWatchFileConfig().getNameReplacement() != null) {
						httpFileUri = fileUri.replaceFirst(
								kafkaWatchFileFilter.getKafkaWatchFileConfig().getNameRegex(),
								kafkaWatchFileFilter.getKafkaWatchFileConfig().getNameReplacement());
					}
					template.send(topic, fileUri, httpFileUri);
					template.flush();

				} else {
					logger.error("Topic is null with dir:{}", file.getAbsolutePath());
				}
			}

		}

	}

	@Override
	public KafkaWatchFileFilter getKafkaWatchFileFilter() {

		return this.kafkaWatchFileFilter;
	}

}
